package cn.moefly.common.component;

import cn.moefly.common.component.support.DelayTask;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.Optional;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


/**
 * 一个统一的延迟任务处理器，由延迟队列和一个消耗线程，一个动态的过期任务执行线程池组成
 */
@Slf4j
@Component
public class DelayTaskUnifiedExecutor {

    public static final String DEFAULT_EXECUTOR_BEAN_NAME = "cn.moefly.delaytask.executor";

    final DelayQueue<DelayTask> delayQueue = new DelayQueue<>();


    /**
     * 关于这俩东西的区别，有必要深入理解下，，晚点回忆然后写下笔记
     * https://blog.csdn.net/Websphere_zxf/article/details/103200011
     * https://cloud.tencent.com/developer/article/1408125
     *
     * ThreadPoolTaskExecutor是对ThreadPoolExecutor进行了封装处理
     */
    ThreadPoolTaskExecutor delayExecutor;

    /**
     * 这个名字好听吗 puppy
     */
    final ExecutorService puppy = Executors.newFixedThreadPool(1);

    @Autowired
    public void setDelayExecutor(@Autowired(required = false) @Qualifier(DEFAULT_EXECUTOR_BEAN_NAME) ThreadPoolTaskExecutor delayExecutor) {
        this.delayExecutor = Optional.ofNullable(delayExecutor).orElseGet(()->{
            var pool = new ThreadPoolTaskExecutor();
            pool.setCorePoolSize(1);
            pool.setMaxPoolSize(Runtime.getRuntime().availableProcessors());
            pool.setQueueCapacity(100);
            pool.setThreadNamePrefix("DelayTask-Default-Pool-");
            pool.initialize();
            return pool;
        });
    }

    @PostConstruct
    private void start(){
        puppy.execute(()->{
            while (!Thread.currentThread().isInterrupted()){
                try{
                    consume();
                }catch (InterruptedException ignore){} catch (Throwable t){
                    log.error("consume error",t);
                }
            }
        });
    }

    private void consume() throws InterruptedException {
        var poll = delayQueue.take();
        if (poll.ifTrulyExpires()){
            delayExecutor.execute(poll::doExpireAction);
        } else {
            poll.updateExpireTime();
            delayQueue.put(poll);
        }
    }

    public void accept(DelayTask task){
        delayQueue.put(task);
    }
}
